package org.example.datastream.test.custom;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class EnterpriseJdbcSink extends RichSinkFunction<RecordLog<Enterprise>> {
    private Logger logger = LoggerFactory.getLogger(EnterpriseJdbcSink.class);

    private String upsertSql = "insert into t_enterprise(id,eid,enterprise_name,logo_url," +
            "oper_name,business_url,department,`type`,status_code,created_time,row_update_time)" +
            "values(null,?,?,?,?,?,?,?,?,?,?)" +
            " ON DUPLICATE KEY UPDATE eid=?, enterprise_name=?, " +
            "logo_url=?, business_url=?, department=?, type=?," +
            "status_code=?,start_date=?,created_time=?,row_update_time=?";

    private String updateSql = "update t_enterprise set enterprise_name=?,logo_url=?," +
            "business_url=?,department=?,type=?,status_code=?,created_time=?,row_update_time=?" +
            " where eid = ?";

    private String deleteSql = "delete from t_enterprise where eid = ?";

    private Connection conn;
    private PreparedStatement ips;
    private PreparedStatement ups;
    private PreparedStatement dps;

    private String proxy;
    private String driver;
    private String host;
    private String port;
    private String db;
    private String username;
    private String password;

    public EnterpriseJdbcSink(String proxy, String driver, String host, String port,
                              String db, String username, String password) {
        this.proxy = proxy;
        this.driver = driver;
        this.host = host;
        this.port = port;
        this.db = db;
        this.username = username;
        this.password = password;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        Class.forName(driver);

        String url = String.format("jdbc:%s://%s:%s/%s?useSSL=false&useUnicode=true&characterEncoding=UTF-8",
                proxy, host, port, db);
        conn = DriverManager.getConnection(url, username, password);
        ips = conn.prepareStatement(upsertSql);
        ups = conn.prepareStatement(updateSql);
        dps = conn.prepareStatement(deleteSql);
    }

    @Override
    public void invoke(RecordLog<Enterprise> value, Context context) throws Exception {
        try {

            String opt = value.getOpt();
            if ("INSERT".equals(opt)) {

                Enterprise ent = value.getRecord();
                ips.setString(1,ent.getEid());
                ips.setString(2,ent.getEnterpriseName());
                ips.setString(3,ent.getLogoUrl());
                ips.setString(4,ent.getOperName());
                ips.setString(5,ent.getBusinessUrl());
                ips.setString(6,ent.getDepartment());
                ips.setInt(7,ent.getType());
                ips.setInt(8,ent.getStatusCode());
                ips.setTimestamp(9,ent.getCreateTime());
                ips.setTimestamp(10,ent.getRowUpdateTime());

                ips.setString(11,ent.getEid());
                ips.setString(12,ent.getEnterpriseName());
                ips.setString(13,ent.getLogoUrl());
                ips.setString(14,ent.getOperName());
                ips.setString(15,ent.getBusinessUrl());
                ips.setString(16,ent.getDepartment());
                ips.setInt(17,ent.getType());
                ips.setInt(18,ent.getStatusCode());
                ips.setTimestamp(19,ent.getCreateTime());
                ips.setTimestamp(20,ent.getRowUpdateTime());
//                logger.warn("ps:{}", ips.toString());
                ips.execute();
            } else if ("UPDATE".equals(opt)) {

                Enterprise ent = value.getRecord();
                ups.setString(1,ent.getEnterpriseName());
                ups.setString(2,ent.getLogoUrl());
                ups.setString(3,ent.getBusinessUrl());
                ups.setString(4,ent.getDepartment());
                ups.setInt(5, ent.getType());
                ups.setInt(6,ent.getStatusCode());
                ups.setTimestamp(7,ent.getCreateTime());
                ups.setTimestamp(8,ent.getRowUpdateTime());
                ups.setString(9, ent.getEid());
//                logger.warn("ps:{}", ups.toString());
                ups.executeUpdate();
            } else if ("DELETE".equals(opt)) {

                dps.setString(1, value.getPkv());
//                logger.warn("ps:{}", dps.toString());
                dps.execute();
            }
        } catch (Exception e) {
            logger.error("error:{}", value);
            throw e;
        }
    }

    @Override
    public void close() throws Exception {
        super.close();

        ips.close();
        dps.close();
        conn.close();
    }
}
